package org.myfram.flink.flinkonjar.common.connector.oracle.source;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.connector.jdbc.JdbcRowDataInputFormat;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.myfram.flink.flinkonjar.domain.entity.OracleSourceTable;

public class OracleSourceBuilder {

    public static DataStreamSource<OracleSourceTable> build(StreamExecutionEnvironment env, String url, String username, String password, String query) {
        JdbcInputFormat inputFormat = JdbcInputFormat.buildJdbcInputFormat()
                .setDrivername("oracle.jdbc.driver.OracleDriver")
                .setDBUrl(url)
                .setUsername(username)
                .setPassword(password)
                .setQuery(query)
                .setRowTypeInfo(TypeInformation.of(OracleSourceTable.class))
                .finish();

        return env.createInput(inputFormat, TypeInformation.of(OracleSourceTable.class));
    }
}